package cn.sciento.transfer.app.service.impl;

import cn.sciento.core.domain.Page;
import cn.sciento.core.exception.CommonException;import io.choerodon.mybatis.pagehelper.domain.PageRequest;
import java.util.List;
import javax.validation.Validator;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import cn.sciento.core.base.BaseConstants;
import cn.sciento.core.util.ValidUtils;
import cn.sciento.transfer.api.dto.ProducerConfigDTO;
import cn.sciento.transfer.app.service.AbstractDataInitService;
import cn.sciento.transfer.app.service.ConsDbConfigService;
import cn.sciento.transfer.app.service.ProducerConfigService;
import cn.sciento.transfer.app.service.TransferRedisService;
import cn.sciento.transfer.domain.entity.ConsDbConfig;
import cn.sciento.transfer.domain.entity.ConsTenantConfig;
import cn.sciento.transfer.domain.entity.ProducerConfig;
import cn.sciento.transfer.domain.repository.ConsDbConfigRepository;
import cn.sciento.transfer.domain.repository.ConsTenantConfigRepository;
import cn.sciento.transfer.domain.repository.EventSendRepository;
import cn.sciento.transfer.domain.repository.ProducerConfigRepository;
import cn.sciento.transfer.infra.datasource.sql.SqlUtils;
import cn.sciento.transfer.infra.rocketmq.config.MqConfigProperties;
import cn.sciento.transfer.infra.rocketmq.service.MqConsumerService;
import cn.sciento.transfer.infra.rocketmq.service.MqProducerService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;

@Service
public class ProducerConfigServiceImpl implements ProducerConfigService {
  private Logger LOGGER = LoggerFactory.getLogger(ProducerConfigServiceImpl.class);
  
  private static final String MULTI_LANG_SUFFIX = "_tl";
  
  private static final String SQL_SPLIT = ";";
  
  public static final String TENANT_ID = "tenant_id";
  
  @Autowired
  private ProducerConfigRepository producerConfigRepository;
  
  @Autowired
  private Validator validator;
  
  @Autowired
  private AsyncTaskExecutor asyncTaskExecutor;
  
  @Autowired
  private ConsTenantConfigRepository consTenantConfigRepository;
  
  @Autowired
  private ConsDbConfigRepository consDbConfigRepository;
  
  @Autowired
  private ConsDbConfigService consDbConfigService;
  
  @Autowired
  private EventSendRepository eventSendRepository;
  
  @Autowired
  private MqConfigProperties mqConfigProperties;
  
  @Autowired
  private MqProducerService mqProducerService;
  
  @Autowired
  private MqConsumerService mqConsumerService;
  
  @Autowired
  private TransferRedisService transferRedisService;
  
  @Autowired
  private AbstractDataInitService dataInitDbServiceImpl;
  
  @Autowired
  private AbstractDataInitService dataInitTenantServiceImpl;
  
  public ProducerConfig createProducerConfig(ProducerConfig producerConfig) {
    if (producerConfig == null) {
      return producerConfig;
    } else {
      producerConfig.init();
      ValidUtils.valid(this.validator, producerConfig);
      producerConfig.validUnique(this.producerConfigRepository);
      if (!SqlUtils.checkTableExists(producerConfig.getInitDsCode(), producerConfig.getServiceCode(), producerConfig.getInitDbCode(), producerConfig.getTableName())) {
        throw new CommonException("error.produce.config..table.not.exist", producerConfig.getTableName(), producerConfig.getInitDbCode());
      } else {
        String tableInitDDL = SqlUtils.getTableInitDDL(producerConfig.getTableName(), producerConfig.getInitDsCode(), producerConfig.getInitDbCode(), producerConfig.getServiceCode());
        Assert.isTrue(StringUtils.isNotEmpty(tableInitDDL), "error.data_invalid");
        producerConfig.setInitDdlSql(tableInitDDL);
        if (BaseConstants.Flag.YES.equals(producerConfig.getTenantFlag()) && !tableInitDDL.contains("tenant_id")) {
          throw new CommonException("error.tenant.not.exist");
        } else {
          String multiLangTableName = producerConfig.getTableName() + "_tl";
          if (SqlUtils.checkTableExists(producerConfig.getInitDsCode(), producerConfig.getServiceCode(), producerConfig.getInitDbCode(), multiLangTableName)) {
            String multiLangTableInitDDL = SqlUtils.getTableInitDDL(multiLangTableName, producerConfig.getInitDsCode(), producerConfig.getInitDbCode(), producerConfig.getServiceCode());
            StringBuilder builder = new StringBuilder();
            builder.append(producerConfig.getInitDdlSql()).append(";").append(multiLangTableInitDDL);
            producerConfig.setInitDdlSql(builder.toString());
          }

          this.createAndAddToRedis(producerConfig);
          this.mqProducerService.createTopic(producerConfig, this.producerConfigRepository);
          return producerConfig;
        }
      }
    }
  }


  @Transactional(
          rollbackFor = {Exception.class}
  )
  ProducerConfig createAndAddToRedis(ProducerConfig producerConfig) {
    this.producerConfigRepository.insertSelective(producerConfig);
    this.transferRedisService.addProducerToRedis(producerConfig);
    return producerConfig;
  }
  
  public ProducerConfig updateProducerConfig(ProducerConfig producerConfig) {
    this.producerConfigRepository.updateOptional(producerConfig, "description", "enabledFlag", "initDdlSql", "initImportDataFlag", "tenantFlag");
    if (BaseConstants.Flag.YES.equals(producerConfig.getTenantFlag()) && 
      !producerConfig.getInitDdlSql().contains("tenant_id"))
      throw new CommonException("error.tenant.not.exist");
    if (BaseConstants.Flag.NO.equals(producerConfig.getEnabledFlag())) {
      this.transferRedisService.removeProducerFromRedis(producerConfig);
    } else {
      this.transferRedisService.addProducerToRedis(producerConfig);
    } 
    return producerConfig;
  }
  
  public Page<ProducerConfig> listProducerConfig(PageRequest pageRequest, ProducerConfigDTO producerConfig) {
    return this.producerConfigRepository.listProducerConfig(pageRequest, producerConfig);
  }
  
  public ConsDbConfig initHistoryData(ConsDbConfig consDbConfig) {
    ProducerConfig producerConfig = (ProducerConfig)this.producerConfigRepository.selectByPrimaryKey(consDbConfig.getProducerConfigId());
    if (!BaseConstants.Flag.YES.equals(producerConfig.getEnabledFlag())) {
      throw new CommonException("error.current_config_disable");
    } else {
      this.asyncTaskExecutor.execute(() -> {
        if (consDbConfig.initByDbDimension()) {
          AbstractDataInitService.DataInitContext dataInitContext = new AbstractDataInitService.DataInitContext(producerConfig, consDbConfig, (ConsTenantConfig)null);
          this.dataInitDbServiceImpl.doInit(dataInitContext);
        } else {
          consDbConfig.getConsTenantConfigList().forEach((item) -> {
            AbstractDataInitService.DataInitContext dataInitContext = new AbstractDataInitService.DataInitContext(producerConfig, consDbConfig, item);
            this.dataInitTenantServiceImpl.doInit(dataInitContext);
          });
        }

      });
      return consDbConfig;
    }
  }
  
  public void initConfigsToRedis() {
    this.transferRedisService.deleteAll();
    ProducerConfig querryProducerConfig = new ProducerConfig();
    querryProducerConfig.setEnabledFlag(BaseConstants.Flag.YES);
    List<ProducerConfig> producerConfigs = this.producerConfigRepository.select(querryProducerConfig);
    if (CollectionUtils.isEmpty(producerConfigs))
      return; 
    for (ProducerConfig producerConfig : producerConfigs) {
      this.transferRedisService.addProducerToRedis(producerConfig);
      ConsDbConfig querryConsDbConfig = new ConsDbConfig();
      querryConsDbConfig.setProducerConfigId(producerConfig.getProducerConfigId());
      List<ConsDbConfig> consDbConfigs = this.consDbConfigRepository.select(querryConsDbConfig);
      if (CollectionUtils.isEmpty(consDbConfigs))
        continue; 
      for (ConsDbConfig consDbConfig : consDbConfigs) {
        if (BaseConstants.Flag.YES.equals(producerConfig.getTenantFlag())) {
          ConsTenantConfig querryConsTenantConfig = new ConsTenantConfig();
          querryConsTenantConfig.setProducerConfigId(producerConfig.getProducerConfigId());
          querryConsTenantConfig.setConsDbConfigId(consDbConfig.getConsDbConfigId());
          List<ConsTenantConfig> consTenantConfigs = this.consTenantConfigRepository.select(querryConsTenantConfig);
          consDbConfig.setConsTenantConfigList(consTenantConfigs);
        } 
        this.transferRedisService.addConsumerToRedis(producerConfig.getServiceCode(), producerConfig.getTableName(), consDbConfig);
      } 
    } 
  }
  
  public void reCreateTopic(Long producerConfigId) {
    ProducerConfig producerConfig = (ProducerConfig)this.producerConfigRepository.selectByPrimaryKey(producerConfigId);
    Assert.notNull(producerConfig, "error.data_invalid");
    this.mqProducerService.createTopic(producerConfig, this.producerConfigRepository);
  }
  
  public ProducerConfig refreshDDL(ProducerConfig producerConfig) {
    if (!SqlUtils.checkTableExists(producerConfig
        .getInitDsCode(), producerConfig
        .getServiceCode(), producerConfig
        .getInitDbCode(), producerConfig
        .getTableName()))
      throw new CommonException("error.produce.config..table.not.exist", producerConfig.getTableName(), producerConfig.getInitDbCode());
    String tableInitDDL = SqlUtils.getTableInitDDL(producerConfig.getTableName(), producerConfig.getInitDsCode(), producerConfig.getInitDbCode(), producerConfig.getServiceCode());
    producerConfig.setInitDdlSql(tableInitDDL);
    String multiLangTableName = producerConfig.getTableName() + "_tl";
    if (SqlUtils.checkTableExists(producerConfig.getInitDsCode(), producerConfig.getServiceCode(), producerConfig.getInitDbCode(), multiLangTableName)) {
      String multiLangTableInitDDL = SqlUtils.getTableInitDDL(multiLangTableName, producerConfig.getInitDsCode(), producerConfig.getInitDbCode(), producerConfig.getServiceCode());
      StringBuilder builder = (new StringBuilder()).append(producerConfig.getInitDdlSql()).append(";").append(multiLangTableInitDDL);
      producerConfig.setInitDdlSql(builder.toString());
    } 
    this.producerConfigRepository.updateOptional(producerConfig, "initDdlSql");
    return null;
  }
  
  @Transactional(rollbackFor = {Exception.class})
  public Integer removeProducerConfig(ProducerConfig producerConfig) {
    this.mqConsumerService.deleteConsumer(producerConfig);
    List<ConsDbConfig> consDbConfigs = this.consDbConfigRepository.select("producerConfigId", producerConfig.getProducerConfigId());
    if (CollectionUtils.isNotEmpty(consDbConfigs))
      consDbConfigs.forEach(item -> this.consDbConfigService.deleteConsDbConfig(item.getConsDbConfigId())); 
    int count = this.producerConfigRepository.deleteByPrimaryKey(producerConfig.getProducerConfigId());
    this.mqProducerService.deleteTopic(producerConfig.getServiceCode(), producerConfig.getTableName());
    this.transferRedisService.removeProducerFromRedis(producerConfig);
    return count;
  }
}
